package com.daviteq.ems.dao.cassandra;

import java.io.Serializable;
import java.lang.reflect.ParameterizedType;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.log4j.Logger;

import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.ConsistencyLevel;
import com.datastax.driver.core.DataType;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import com.datastax.driver.core.querybuilder.Select;
import com.datastax.driver.core.querybuilder.Select.Builder;
import com.datastax.driver.core.querybuilder.Update;
import com.datastax.driver.core.querybuilder.Update.Assignments;
import com.datastax.driver.core.querybuilder.Update.Where;
import com.datastax.driver.mapping.EntityFieldMetaData;
import com.datastax.driver.mapping.EntityTypeMetadata;
import com.datastax.driver.mapping.EntityTypeParser;
import com.datastax.driver.mapping.PrimaryKeyMetadata;
import com.daviteq.ems.dao.IBasicDAO;

/**
 * @author thanhthi.tran
 * @date Mar 28, 2014
 * @since 0.0.1
 * @param <T> Entity
 * @param <K> Key
 */

public class BasicDAO<T, K extends Serializable> implements
        IBasicDAO<T, K> {

    private final static Logger log = Logger.getLogger(BasicDAO.class);
    
	protected final static int MAX_RECORD = 1000;
	
	protected SessionFatory sf = SessionFatory.getInstance();

    @SuppressWarnings("unchecked")
    protected Class<T> classT = (Class<T>) ((ParameterizedType) getClass()
            .getGenericSuperclass()).getActualTypeArguments()[0];

    protected EntityTypeMetadata emeta = EntityTypeParser
            .getEntityMetadata(classT);

    public BasicDAO() {
    }

    @Override
    public T findById(K id) {
        return (T) sf.getMappingSession().get(classT, id);
    }

    @Override
    public boolean remove(K id) {
        try {
            if (exists(id)) {
                sf.getMappingSession().delete(findById(id));
                return true;
            } 
        } catch (Exception ex) {
            log.error(ex);
        }
        
        return false;
        
    }

    @Override
    public boolean persist(T t) {
        try {
            sf.getMappingSession().save(t);
            return true;
        } catch (Exception ex) {
            log.error(ex);
        }
        
        return false;
    }

    @Override
    public boolean exists(K id) {
        if (null != findById(id)) {
            return true;
        }
        return false;
    }

    @Override
    public List<T> findBySecondaryIndex(String colName, String colValue, String... colsRet) {
        
        Map<String, Object> cols = new HashMap<String, Object>();
        cols.put(colName, colValue);
        
        return findBySecondaryIndexes(cols, colsRet);
        
    }

    @Override
    public boolean updateMultiColsById(Map<String, Object> cols, String pk, String val) {

        if (null == cols
                || null == pk
                || null == val) {
            return false;
        }
        
        try {
            
            List<String> listColName = new ArrayList<String>();
        
            List<Object> listBindVal = new ArrayList<Object>();
        
            for (Entry<String, Object> entry : cols.entrySet()) {
                listColName.add(entry.getKey());
                listBindVal.add(entry.getValue());
            }
        
            /** add value of primary key **/
            listBindVal.add(val);
            
            String query = buildUpdateForColumns(listColName, pk);
            PreparedStatement pstmt = sf.getSession().prepare(query);
            BoundStatement bstmt = pstmt.bind(listBindVal.toArray());
            
            sf.getSession().execute(bstmt);
            
            return true;
            
        } catch (Exception ex) {
            log.error(ex);
        }
        
        return false;
        

    }

    @Override
    public List<T> findBySecondaryIndexes(Map<String, Object> cols, String... colsRet) {

        List<String> listColName = new ArrayList<String>();
        List<Object> listColVal = new ArrayList<Object>();

        for (Entry<String, Object> entry : cols.entrySet()) {
            listColName.add(entry.getKey());
            listColVal.add(entry.getValue());
        }

        String query = buildSelectForColumns(listColName, colsRet);
        
        PreparedStatement pstmt = sf.getSession().prepare(query);
        BoundStatement bstmt = pstmt.bind(listColVal.toArray());
        ResultSet ret = sf.getSession().execute(bstmt);

        // List<T> list = sf.getMappingSession().getFromResultSet(classT, ret);
        // if(list.size() <= 0) {
        // return null;
        // }
        
        return getFromResultSet(classT, ret);
    }

    protected String buildUpdateForColumns(List<String> cols, String pk) {

        /* build UPDATE */
        Update update = QueryBuilder.update(sf.getKeyspace(),
                emeta.getTableName());

        /* build SET */
        Assignments assignments = update.with();
        for (String col : cols) {
            assignments.and(QueryBuilder.set(col, "?"));
        }

        /* build WHERE */
        Where where = update.where();
        where.and(QueryBuilder.eq(pk, "?"));

        return update.getQueryString();

    }

    protected String buildSelectForColumns(List<String> cols, String... colsRet) {
        
        Builder builder = QueryBuilder.select().all();

        if (colsRet.length > 0) {
            builder = null;
            builder = QueryBuilder.select(colsRet);
        }
        
        Select select = builder.from(sf.getKeyspace(), emeta.getTableName());

        for (String col : cols) {
            select.where().and(QueryBuilder.eq(col, "?"));
        }

        select.allowFiltering();
        
        select.setConsistencyLevel(ConsistencyLevel.QUORUM);
        
        return select.getQueryString();

    }

	@Override
	public List<T> findAll(int size,  String... colsRet) {
		
		if (size > MAX_RECORD) {
			size = MAX_RECORD;
		}
		
		Builder builder = QueryBuilder.select().all();

        if (null == colsRet) {
            builder = null;
            builder = QueryBuilder.select(colsRet);
        }
		
        Select select = builder.from(sf.getKeyspace(), emeta.getTableName()).limit(size);

		ResultSet rs = sf.getSession().execute(select.getQueryString());
		
		return getFromResultSet(classT, rs);
		
	}

    @Override
    public boolean existsSecondaryIndex(String colName, String colValue) {
        
        List<T> entity = findBySecondaryIndex(colName, colValue);
        if (null!=entity
                && entity.size() > 0 ) {
            return true;
        }
        
        return false;
    }
    
    private Object getValueFromRow(Row row, EntityFieldMetaData field) {
        Object value = null;
        try {
            DataType.Name dataType = field.getDataType();
            switch (dataType) {
                case BLOB:
                    value = row.getBytes(field.getColumnName());
                    break;
                case BOOLEAN:
                    value = row.getBool(field.getColumnName());
                    break;
                case TEXT:
                    value = row.getString(field.getColumnName());
                    break;
                case TIMESTAMP:
                    value = row.getDate(field.getColumnName());
                    break;
                case UUID:
                    value = row.getUUID(field.getColumnName());
                    break;
                case INT:
                    value = row.getInt(field.getColumnName());
                    break;
                case DOUBLE:
                    value = row.getDouble(field.getColumnName());
                    break;
                case BIGINT:
                    value = row.getLong(field.getColumnName());
                    break;
                case DECIMAL:
                    value = row.getDecimal(field.getColumnName());
                    break;
                case VARINT:
                    value = row.getVarint(field.getColumnName());
                    break;
                case FLOAT:
                    value = row.getFloat(field.getColumnName());
                    break;                      
                case MAP:
                    value = row.getMap(field.getColumnName(), Object.class, Object.class);
                    break;
                case LIST:
                    value = row.getList(field.getColumnName(), Object.class);
                    break;
                case SET:
                    value = row.getSet(field.getColumnName(), Object.class);
                    break;
                default:
                    break;
            }
        } catch (Exception ex) {
        }
        return value;
    }
    
    protected <T> List<T> getFromResultSet(Class<T> clazz, ResultSet rs) {
        List<T> result = new ArrayList<T>();
        EntityTypeMetadata entityMetadata = EntityTypeParser.getEntityMetadata(clazz);
        for (Row row: rs.all()) {
            T entity = null;
            Object primaryKey = null;
            Object partitionKey = null;
            
            try {
                entity = clazz.newInstance();
                PrimaryKeyMetadata pkmeta = entityMetadata.getPrimaryKeyMetadata();
                if (pkmeta.isCompound()) {
                    EntityFieldMetaData pkField = pkmeta.getOwnField();
                    primaryKey = pkField.getType().newInstance();
                    pkField.setValue(entity, primaryKey);
                    if (pkmeta.hasPartitionKey()) {
                        PrimaryKeyMetadata partmeta = pkmeta.getPartitionKey();
                        EntityFieldMetaData partField = partmeta.getOwnField(); 
                        partitionKey = partField.getType().newInstance();
                        partField.setValue(primaryKey, partitionKey);
                    }
                }
            } catch (Exception ex) {
                return null;
            }
            
            for (EntityFieldMetaData field: entityMetadata.getFields()) {
                Object value = getValueFromRow(row, field);
                try {   
                    if (value != null) {
                        if (field.isPartition()) {
                            field.setValue(partitionKey, value);    
                        } else if (field.isPrimary()) {
                            field.setValue(primaryKey, value);
                        } else {
                            field.setValue(entity, value);
                        }
                        
                    }
                } catch (Exception ex) {
                    // field is not properly mapped, log debug and skip the error
                }
            }
            result.add(entity);
        }
        return result;
    }

}
